package com.fyt.reachgo.consummer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fyt.reachgo.config.RabbitMQConfig;
import com.fyt.reachgo.domain.CommentLike;
import com.fyt.reachgo.domain.message.CommentLikeMessage;
import com.fyt.reachgo.domain.TravelComment;
import com.fyt.reachgo.mapper.CommentLikeMapper;
import com.fyt.reachgo.mapper.TravelCommentMapper;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import static com.fyt.reachgo.utils.RedisConstants.CACHE_TRAVEL_GUIDE_COMMENTS_KEY;

@Component
@Slf4j
@RequiredArgsConstructor
public class CommentMessageConsumer {
    private static final String LIKE_IDEMPOTENT_KEY_PREFIX = "like:";
    private static final String COMMENT_CACHE_KEY_PREFIX = "travel_guide_comments:";

    private final TravelCommentMapper commentMapper;
    private final CommentLikeMapper commentLikeMapper;
    private final StringRedisTemplate stringRedisTemplate;

    @RabbitListener(
            queues = RabbitMQConfig.COMMENT_ADD_QUEUE,
            ackMode = "MANUAL",
            concurrency = "3-5" // 根据实际情况调整并发数
    )
    public void handleCommentAdd(TravelComment comment, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            log.info("收到新评论消息, 评论ID: {}, 指南ID: {}", comment.getId(), comment.getTravelGuideId());

            // 清理相关缓存
            clearCommentCache(comment.getTravelGuideId());

            // 手动确认消息
            channel.basicAck(tag, false);
            log.info("评论添加消息处理完成, 评论ID: {}", comment.getId());
        } catch (Exception e) {
            log.error("处理评论添加消息失败, 评论ID: " + comment.getId(), e);
            try {
                // 处理失败，拒绝消息并重新入队
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                log.error("消息NACK失败, 评论ID: " + comment.getId(), ex);
            }
        }
    }

    @RabbitListener(
            queues = RabbitMQConfig.COMMENT_LIKE_QUEUE,
            ackMode = "MANUAL",
            concurrency = "5-10" // 点赞操作可以更高并发
    )
    @Transactional(rollbackFor = Exception.class)
    public void handleCommentLike(CommentLikeMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            log.info("收到评论点赞消息, 用户ID: {}, 评论ID: {}, 操作: {}",
                    message.getUserId(), message.getCommentId(), message.isLike() ? "点赞" : "取消点赞");

            // 幂等性检查
            if (!checkAndSetIdempotent(message)) {
                channel.basicAck(tag, false);
                return;
            }

            // 处理点赞/取消点赞
            if (message.isLike()) {
                processLike(message);
            } else {
                processUnlike(message);
            }

            // 清理相关缓存
            clearCommentCacheByCommentId(message.getCommentId());

            // 确认消息
            channel.basicAck(tag, false);
            log.info("评论点赞消息处理完成, 用户ID: {}, 评论ID: {}", message.getUserId(), message.getCommentId());
        } catch (Exception e) {
            log.error("处理评论点赞消息失败, 用户ID: {}, 评论ID: {}", message.getUserId(), message.getCommentId(), e);
            try {
                // 处理失败，拒绝消息并重新入队
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                log.error("消息NACK失败, 用户ID: {}, 评论ID: {}", message.getUserId(), message.getCommentId(), ex);
            }
            throw new RuntimeException(e); // 触发事务回滚
        }
    }

    private boolean checkAndSetIdempotent(CommentLikeMessage message) {
        String idempotentKey = LIKE_IDEMPOTENT_KEY_PREFIX + message.getUserId() + ":" + message.getCommentId();
        return Boolean.TRUE.equals(stringRedisTemplate.opsForValue()
                .setIfAbsent(idempotentKey, "1", 5, TimeUnit.MINUTES));
    }

    private void processLike(CommentLikeMessage message) {
        // 检查是否已点赞
        if (commentLikeMapper.exists(new LambdaQueryWrapper<CommentLike>()
                .eq(CommentLike::getUserId, message.getUserId())
                .eq(CommentLike::getCommentId, message.getCommentId()))) {
            log.warn("用户已点赞过该评论, 用户ID: {}, 评论ID: {}", message.getUserId(), message.getCommentId());
            return;
        }

        // 新增点赞记录
        CommentLike like = new CommentLike();
        like.setCommentId(message.getCommentId());
        like.setUserId(message.getUserId());
        commentLikeMapper.insert(like);

        // 更新点赞数
        commentMapper.incrementLikeCount(message.getCommentId(), 1);
    }

    private void processUnlike(CommentLikeMessage message) {
        // 检查是否存在点赞记录
        if (!commentLikeMapper.exists(new LambdaQueryWrapper<CommentLike>()
                .eq(CommentLike::getUserId, message.getUserId())
                .eq(CommentLike::getCommentId, message.getCommentId()))) {
            log.warn("用户未点赞过该评论, 无法取消, 用户ID: {}, 评论ID: {}", message.getUserId(), message.getCommentId());
            return;
        }

        // 删除点赞记录
        commentLikeMapper.delete(new LambdaQueryWrapper<CommentLike>()
                .eq(CommentLike::getUserId, message.getUserId())
                .eq(CommentLike::getCommentId, message.getCommentId()));

        // 更新点赞数
        commentMapper.incrementLikeCount(message.getCommentId(), -1);
    }

    private void clearCommentCache(Integer guideId) {
        String pattern = COMMENT_CACHE_KEY_PREFIX + guideId + ":*";
        Set<String> keys = stringRedisTemplate.keys(pattern);
        if (keys != null && !keys.isEmpty()) {
            stringRedisTemplate.delete(keys);
            log.debug("清理评论缓存, 指南ID: {}, 清理数量: {}", guideId, keys.size());
        }
    }

    private void clearCommentCacheByCommentId(Long commentId) {
        TravelComment comment = commentMapper.selectById(commentId);
        if (comment != null) {
            clearCommentCache(comment.getTravelGuideId());
        }
    }
}